package studio.raptor.databus.client;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.ActiveMQConnectionFactory;
import studio.raptor.databus.client.config.BrokerGroup;
import studio.raptor.databus.command.EventProtoBuf;

import javax.jms.*;
import java.util.List;

/**
 * 队列消息生产者，每个生产者只能发送消息到一个指定的主题
 */
public final class DatabusProducer {

  private static Logger log = LoggerFactory.getLogger(DatabusProducer.class);
  private DatabusConfig databusConfig;
  private DatabusProducerBuilder databusProducerBuilder;

  private HashMap<Integer/*brokerGroupId*/, Connection> brokerGroupIdConMap =
      new HashMap<Integer/*brokerGroupId*/, Connection>(5);
  private HashMap<Integer/*brokerGroupId*/, Session> brokerGroupIdSessMap =
      new HashMap<Integer/*brokerGroupId*/, Session>(5);
  private HashMap<Integer/*brokerGroupId*/, MessageProducer> brokerGroupIdProMap =
      new HashMap<Integer/*brokerGroupId*/, MessageProducer>(5);

  /**
   * 在使用sendWithNoCommit方法时正在发送中的事件类型
   */
  private String currentSendingEventNameKindId = null;

  private static final String DELIVERTIME_FORMAT_REGEXP = "(19[0-9]{2})|(2[0-9]{3})-(0[1-9]|1[012])-(0[1-9]|[12][0-9]|3[01]) ([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]";

  private static final Pattern PATTERN = Pattern.compile(DELIVERTIME_FORMAT_REGEXP);
  //private HashMap<Integer/*eventKindId*/, Integer/*brokerGroupId*/> eventKindIdGrpIdMap = new HashMap<Integer/*eventKindId*/, Integer/*brokerGroupId*/>();

  private List<BrokerGroup> brokerGroups = new ArrayList<BrokerGroup>();

  /**
   * 构造方法,对包内其他类开放
   */
  DatabusProducer(DatabusConfig databusConfig) {
    this.databusConfig = databusConfig;
  }

  /**
   * 通过DatabusConfig实例和DatabusProducerBuilder实例生成消息发送者
   *
   * @param databusConfig databus配置实例
   * @param producerBuilder producerBuilder配置实例
   * @param brokerGroups brokergroup配置信息
   */
  DatabusProducer(DatabusConfig databusConfig, DatabusProducerBuilder producerBuilder,
      List<BrokerGroup> brokerGroups) {
    this.databusConfig = databusConfig;
    this.databusProducerBuilder = producerBuilder;
    //this.eventKindIdGrpIdMap = eventKindIdGrpIdMap;
    this.brokerGroups = brokerGroups;
  }

  /**
   * 批量发送消息事件.<br>
   * 1、事件集合中事件字段违反约束时，全部事件发送失败。建议在批量发送事件之前自行校验<br>
   * 2、事件集合中事件未允许发送授权时，全部事件发送失败。建议在批量发送事件之前自行校验<br>
   * 3、事件集合中必须保证所有的事件eventKindId相同。
   *
   * @param eventList 事件列表
   */
  public void send(List<EventProtoBuf.Event> eventList) throws JMSException {
    if (eventList == null) {
      throw new JMSException("事件不可为空");
    }
    if (eventList.size() == 0) {
      return;
    }
    Integer eventKindIdTemp = eventList.get(0).getEventBasicData().getKindId();
    String eventNameTemp = eventList.get(0).getEventBasicData().getName();

    //检查当前集合是否存在不同eventKindId的事件
    for (int i = 0; i < eventList.size(); i++) {
      if (eventList.get(i).getEventBasicData().getKindId() != eventKindIdTemp) {
        throw new JMSException("批量发送事件不可发送不同种类的事件！");
      }
      //是否有权限发送事件
      if (!sendAuthorizationCheck(eventList.get(i))) {
        log.info(
            "您无权发送事件:[" + eventList.get(i).getEventBasicData().getName() + "-" + eventList.get(i)
                .getEventBasicData().getKindId() + "],若需发送请先申请");
        throw new JMSException(
            "您无权发送事件:[" + eventList.get(i).getEventBasicData().getName() + "-" + eventList.get(i)
                .getEventBasicData().getKindId() + "],若需发送请先申请");
      }
      String validResult = validEvent(eventList.get(i));
      if(!Objects.equals("",validResult)){
        throw new JMSException(validResult);
      }
    }
    //相同类型的事件，使用相同的生产者发送消息
    Integer brokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub().get(eventNameTemp+"_"+eventKindIdTemp);
    if (!brokerGroupIdProMap.containsKey(brokerGroupId) || !brokerGroupIdSessMap
        .containsKey(brokerGroupId)) {
      //若不存在，需要对其进行初始化
      initProducer(brokerGroupId, brokerGroupIdProMap,
          brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);
    }

    for (int i = 0; i < eventList.size(); i++) {
      BytesMessage bytesMessage = this.brokerGroupIdSessMap.get(brokerGroupId).createBytesMessage();
      bytesMessage.writeBytes(eventList.get(i).toByteArray());

      String eventName = eventList.get(i).getEventBasicData().getName();
      Integer eventKindId = eventList.get(i).getEventBasicData().getKindId();
      String topicSalt = TopicSalt.createSalt(eventName, eventKindId);
      String virtualTopic =
          DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + topicSalt + "_" + eventKindId
              .intValue();

      Destination destination = this.brokerGroupIdSessMap.get(brokerGroupId)
          .createTopic(virtualTopic);
      this.brokerGroupIdProMap.get(brokerGroupId).send(destination, bytesMessage);
    }

    this.brokerGroupIdSessMap.get(brokerGroupId).commit();
  }

  /**
   * 发送消息事件
   *
   * @param event 事件
   */
  public void send(EventProtoBuf.Event event) throws JMSException {

    String validResult = validEvent(event);
    if(!Objects.equals("",validResult)){
      throw new JMSException(validResult);
    }

    Integer eventKindId = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();

    //获取当前事件对应的brokerGroupId
    Integer brokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub().get(eventName+"_"+eventKindId);

    initProducer(brokerGroupId, brokerGroupIdProMap,
        brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);

    BytesMessage bytesMessage = this.brokerGroupIdSessMap.get(brokerGroupId).createBytesMessage();
    bytesMessage.writeBytes(event.toByteArray());

    Destination destination =
        createDestination(this.brokerGroupIdSessMap.get(brokerGroupId),eventName,eventKindId);

    reallySend(this.brokerGroupIdProMap.get(brokerGroupId),destination,bytesMessage);
    reallyCommit(this.brokerGroupIdSessMap.get(brokerGroupId));

  }

  /**
   * 真正用于发送消息的方法
   * @param producer 消息生产者
   * @param destination 消息目的地
   * @param message 消息载体对象
   * @throws JMSException
   */
  private void reallySend(MessageProducer producer,Destination destination,Message message)
      throws JMSException {
    producer.send(destination,message);
  }

  /**
   * 会话提交方法
   * @param session 会话
   * @throws JMSException
   */
  private void reallyCommit(Session session) throws JMSException {
    session.commit();
  }

  /**
   * 发送事件但不进行提交(提交之前所发送的事件必须为相同类型)
   * <p>与方法<code>commit()</code>或<code>rollback</code>配合使用</p>
   * <p>此方法在执行commit或者rollback之前,事件种类ID必须一致，否则强制回滚</p>
   * @param event 待发送的事件
   */
  public void sendWithNoCommit(EventProtoBuf.Event event) throws JMSException {
    String validResult = validEvent(event);
    if(!Objects.equals("",validResult)){
      throw new JMSException(validResult);
    }

    Integer eventKindId = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();

    //获取当前事件对应的brokerGroupId
    Integer brokerGroupId =
        this.databusConfig.getBizPubSubEventInfo().getPub().get(eventName+"_"+eventKindId);

    //校验发送的事件是否有不同类型
    if( Objects.equals(null,this.currentSendingEventNameKindId) ){
      //当前发送的消息类型为空时赋值，表示开始发送当前类型事件
      this.currentSendingEventNameKindId = eventName+"_"+eventKindId;
    }else{
      //若发现发送的事件不一致，回滚
      if( !this.currentSendingEventNameKindId.equals(eventName+"_"+eventKindId) ){
        Integer preBrokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub().get(
            currentSendingEventNameKindId);
        this.brokerGroupIdSessMap.get(preBrokerGroupId).rollback();
        this.currentSendingEventNameKindId = null;
        throw new JMSException("使用事务性生产者发送消息，同一次事务中只能发送同一种消息类型");
      }
    }


    initProducer(brokerGroupId, brokerGroupIdProMap,
        brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);

    BytesMessage bytesMessage = this.brokerGroupIdSessMap.get(brokerGroupId).createBytesMessage();
    bytesMessage.writeBytes(event.toByteArray());

    Destination destination =
        createDestination(this.brokerGroupIdSessMap.get(brokerGroupId),eventName,eventKindId);

    reallySend(this.brokerGroupIdProMap.get(brokerGroupId),destination,bytesMessage);

  }

  /**
   * 将已经发送的事件进行提交操作
   * @throws JMSException
   */
  public void commit() throws JMSException {
    if( this.currentSendingEventNameKindId != null ){
      Integer currentSendBrokerGroupId =
          this.databusConfig.getBizPubSubEventInfo().getPub().get(currentSendingEventNameKindId);
      reallyCommit(this.brokerGroupIdSessMap.get(currentSendBrokerGroupId));
      //重置当前生产者可以发送任何消息种类
      this.currentSendingEventNameKindId = null;
    }
  }


  /**
   * 生产者回滚
   * @throws JMSException
   */
  public void rollback() throws JMSException {

    if( this.currentSendingEventNameKindId != null ){
      Integer currentSendBrokerGroupId =
          this.databusConfig.getBizPubSubEventInfo().getPub().get(currentSendingEventNameKindId);
      this.brokerGroupIdSessMap.get(currentSendBrokerGroupId).rollback();
      //重置当前生产者可以发送任何消息种类
      this.currentSendingEventNameKindId = null;
    }

  }

  /**
   * 校验当前生产者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   * @param event 事件对象实例
   * @return 是否授权
   * @deprecated As of databus-client version 0.1.3-SNAPSHOT,
   * replaced by <code>checkSendAuthorization(EventProtoBuf.Event event)</code>.
   */
  @Deprecated
  public boolean sendAuthorizationCheck(EventProtoBuf.Event event) {
    Integer eventSeq = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();
    return sendAuthorizationCheck(eventSeq, eventName);
  }



  /**
   * 校验当前生产者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   * @param eventKindId 事件种类Id
   * @param eventName 事件名称
   * @return
   * @deprecated As of databus-client version 0.1.3-SNAPSHOT,
   * replaced by <code>checkSendAuthorization(Integer eventKindId,String eventName)</code>.
   */
  @Deprecated
  public boolean sendAuthorizationCheck(Integer eventKindId, String eventName) {

    return this.databusConfig.getBizPubSubEventInfo().getPub().containsKey(eventName+"_"+eventKindId);
  }

  /**
   * 校验消费者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   * @param eventKindId 事件类型ID
   * @param eventName 事件名称
   * @return 是否有授权
   */
  public boolean checkSendAuthorization(Integer eventKindId, String eventName){
    return this.databusConfig.getBizPubSubEventInfo().getPub().containsKey(eventName+"_"+eventKindId);
  }

  /**
   * 校验消费者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   * @param event 事件对象实例
   * @return 是否有授权
   */
  public boolean checkSendAuthorization(EventProtoBuf.Event event){
    Integer eventSeq = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();
    return checkSendAuthorization(eventSeq, eventName);
  }

  /**
   * 关闭所有生产者及其对应的会话和连接
   */
  public void close() {
    try {
      //关闭所有的生产者
      Set<Entry<Integer, MessageProducer>> producers = this.brokerGroupIdProMap.entrySet();
      Iterator<Entry<Integer, MessageProducer>> producerIt = producers.iterator();
      while (producerIt.hasNext()) {
        MessageProducer producer = producerIt.next().getValue();
        if (producer != null) {
          producer.close();
        }
      }
      //关闭所有的会话
      Set<Entry<Integer, Session>> sessions = this.brokerGroupIdSessMap.entrySet();
      Iterator<Entry<Integer, Session>> sessionIt = sessions.iterator();
      while (sessionIt.hasNext()) {
        Session session = sessionIt.next().getValue();
        if (session != null) {
          session.close();
        }
      }
      // 关闭所有的连接
      Set<Entry<Integer, Connection>> connections = this.brokerGroupIdConMap.entrySet();
      Iterator<Entry<Integer, Connection>> connectionIt = connections.iterator();
      while (connectionIt.hasNext()) {
        Connection connection = connectionIt.next().getValue();
        if (connection != null) {
          connection.close();
        }
      }
    } catch (JMSException e) {
      log.error(e.getMessage(), e);
    }
  }


  /**
   * 创建指定发送消息到指定Broker的消费者,若已创建则忽略
   *
   * @param brokerGroupId broker组ID
   * @param brokerGroupIdProMap 存放brokerGroupId和生产者的map
   * @param brokerGroupIdSessMap 存放brokerGroupId和会话的map
   * @param brokerGroupIdConMap 存放brokerGroupId和连接的map
   * @param producerBuilder 包含了生产者配置的实例
   */
  private void initProducer(Integer brokerGroupId,
      HashMap<Integer/*brokerGroupId*/, MessageProducer> brokerGroupIdProMap,
      HashMap<Integer/*brokerGroupId*/, Session> brokerGroupIdSessMap,
      HashMap<Integer/*brokerGroupId*/, Connection> brokerGroupIdConMap,
      DatabusProducerBuilder producerBuilder) throws JMSException {

    //若指定的生产者及其对应的会话和连接已经存在，直接返回,不再进行初始化
    if (brokerGroupIdProMap.containsKey(brokerGroupId) && brokerGroupIdSessMap
        .containsKey(brokerGroupId)) {
      return ;
    }

    String userName = "";
    String password = "";
    String brokerURL = DatabusConfig.LOAD_CONFIG_FAILURE;
    //依据brokerGroupId选择配置的url
    for (BrokerGroup brokerGroup : this.brokerGroups) {
      if (brokerGroupId.intValue() == brokerGroup.getBrokerGroupId()) {
        userName = brokerGroup.getUserName();
        password = brokerGroup.getPassword();
        brokerURL = brokerGroup.getBrokerURL();
        break;
      }
    }
    if(DatabusConfig.LOAD_CONFIG_FAILURE.equals(brokerURL)){
      throw new JMSException("当前事件消息未配置发送目的地或配置了错误的目的地");
    }
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
    connectionFactory.setUserName(userName);
    connectionFactory.setPassword(password);
    String customizedBrokerURL = producerBuilder.transform2URLConfig();
    connectionFactory.setBrokerURL(brokerURL + "?" + customizedBrokerURL);

    // 配置生产者客户端中的连接、会话
    Connection connection = connectionFactory.createConnection();
    connection.start();
    Session session = connection
        .createSession(producerBuilder.isTransactedSession(), producerBuilder.getAcknowledgeMode());
    MessageProducer producer = session.createProducer(null);
    producer.setTimeToLive(producerBuilder.getMsgTimeToLive()*24*3600*1000);
    brokerGroupIdConMap.put(brokerGroupId, connection);
    brokerGroupIdSessMap.put(brokerGroupId, session);
    brokerGroupIdProMap.put(brokerGroupId, producer);

  }

  /**
   * 校验事件，校验是否有权限发送；校验对象字段约束
   * @param event 事件对象
   * @return 校验结果，空字符串表示校验通过
   */
  public String validEvent(EventProtoBuf.Event event) {
    String validStr = "";
    if (event == null) {
      validStr ="事件不可为空";
      return validStr;
    }

    //是否有权限发送事件
    if (!checkSendAuthorization(event)) {
      validStr = "您无权发送事件:[" + event.getEventBasicData().getName() + "-" + event.getEventBasicData()
          .getKindId() + "],若需发送请先申请";
      return validStr;
    }
    //字段约束校验
    // 业务数据字段的校验
    EventProtoBuf.EventBusinessData businessData = event.getEventBusinessData();
    if( !Objects.equals(null,businessData)){
        if( Objects.equals(null,businessData.getContent()) || Objects.equals("",businessData.getContent())){
          validStr = "EventBusinessData对象数据携带方式为自身携带，content字段不可为空";
          return validStr;
        }
    }else{
      validStr = "Event中EventBusinessData对象不可为空";
      return validStr;
    }

    //基础数据字段校验
    EventProtoBuf.EventBasicData basicData = event.getEventBasicData();
    if( !Objects.equals(null,basicData)){
      if( Objects.equals(null,basicData.getSource()) || Objects.equals("",basicData.getSource())){
        validStr = "EventBasicData中content字段不可为空";
        return validStr;
      }
      if( Objects.equals(null,basicData.getKindId()) ){
        validStr = "EventBasicData中kindId字段不可为空";
        return validStr;
      }
      if( Objects.equals(null,basicData.getName()) || Objects.equals("",basicData.getName())){
        validStr = "EventBasicData中name字段不可为空";
        return validStr;
      }
      if( Objects.equals(null,basicData.getDeliverTime()) || Objects.equals("",basicData.getDeliverTime())){
        validStr = "EventBasicData中deliverTime字段不可为空";
        return validStr;
      }
      //投递事件的格式和有效性分开校验

      Matcher matcher = PATTERN.matcher(basicData.getDeliverTime());
      if( matcher.matches() ){
        String date = basicData.getDeliverTime().split("[ ]")[0];
        int month = Integer.valueOf(date.split("[-]")[1]);
        if (month == 2) {
          int year = Integer.valueOf(date.split("[-]")[0]);
          Calendar cal = Calendar.getInstance();
          //计算得到2月份的最后一天
          cal.set(year, 1, 1);
          int lastDay = cal.getActualMaximum(Calendar.DAY_OF_MONTH);
          if ( lastDay < Integer.valueOf(date.split("[-]")[2]) ){
            return "EventBasicData中deliverTime字段:[ "+basicData.getDeliverTime()+" ] 不合法" ;
          }
        }
      }else{
        return "EventBasicData中deliverTime字段:[ "+basicData.getDeliverTime()+" ] 不满足 YYYY-MM-DD HH24:MI:SS格式" ;

      }

    }else{
      validStr = "Event中EventBasicData对象不可为空";
      return validStr;
    }

    //元数据字段校验
    EventProtoBuf.EventMetaData metaData = event.getEventMetaData();
    if( !Objects.equals(null,metaData)){
      if( Objects.equals(null,metaData.getBusinessNo()) || Objects.equals("",metaData.getBusinessNo())){
        validStr = "EventMetaData对象中businessNo字段不可为空";
        return validStr;
      }
      if( Objects.equals(null,metaData.getRegionId())){
        validStr = "EventMetaData对象中regionId字段不可为空";
        return validStr;
      }
    }else{
      validStr = "Event中EventMetaData对象不可为空";
      return validStr;
    }
    return validStr ;

  }

  /**
   * 创建 Destination
   * @param session 会话
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @return 消息目的地destination
   * @throws JMSException
   */
  private Destination createDestination(Session session,String eventName,Integer eventKindId)
      throws JMSException {
    String topicSalt = TopicSalt.createSalt(eventName, eventKindId);
    String virtualTopic =
        DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + topicSalt + "_" + eventKindId
            .intValue();
    Destination destination = session.createTopic(virtualTopic);
    return  destination;
  }
}
